IF OBJECT_ID('ssf.ptx_run_task_group') IS NOT NULL
	DROP PROCEDURE ssf.ptx_run_task_group
GO

CREATE PROCEDURE ssf.ptx_run_task_group
	@group_type NVARCHAR(128) = N'DEFAULT'
	, @group_name NVARCHAR(128)
	, @max_threads SMALLINT = 1
	, @log_directory NVARCHAR(200) = NULL
	, @loop_delay TIME(1) = NULL
	, @task_group_run_id NVARCHAR(256) = NULL OUTPUT
AS
/*
<object_header>

<object_summary>
	<purpose><![CDATA[

	]]></purpose>
</object_summary>

</object_header>
*/

SET NOCOUNT ON

-- ============
/*<code_comment><![CDATA[declare logging variables]]></code_comment>*/
DECLARE	@log_id BIGINT
			, @task NVARCHAR(512)
			, @task_description NVARCHAR(2048)
			, @target_object NVARCHAR(512)
			, @source_object NVARCHAR(512) = OBJECT_SCHEMA_NAME(@@PROCID) + N'.' + OBJECT_NAME(@@PROCID)
-- ============

-- ============
/*<code_comment><![CDATA[declare dynamic sql variables]]></code_comment>*/
DECLARE	@proc_error INT
			, @sql_error INT
			, @error_message NVARCHAR(2048)
			, @error_line INT
			, @row_count BIGINT
			, @nsql NVARCHAR(MAX)
			, @crlf CHAR(2) = CHAR(13) + CHAR(10)
			, @quote CHAR(1) = CHAR(39)
			, @tab CHAR(1) = CHAR(9)
-- ============

-- ============
/*<code_comment><![CDATA[declare local variables]]></code_comment>*/
DECLARE	@task_group_id INT
			, @task_group_log_id INT
			, @task_count INT
			, @task_group_run_date DATETIME
			, @gbl_name_max_threads NVARCHAR(128)
			, @gbl_value_max_threads INT
			, @running_max_threads INT
			, @loop_delay_string VARCHAR(12)
			, @job_category NVARCHAR(128)
			, @job_id UNIQUEIDENTIFIER
			, @job_name NVARCHAR(128)
			, @task_id INT
			, @task_database_name NVARCHAR(128)
			, @task_procedure_name NVARCHAR(256)
			, @task_parameter_list NVARCHAR(MAX)
			, @semaphore_name NVARCHAR(128)
			, @connections_released INT
			, @connections_acquired INT
			, @log_file_path NVARCHAR(200)
			, @job_run_datetime DATETIME
			, @job_status INT
			, @failed_fl BIT
			, @cursor_completed CURSOR
			, @cursor_eligible CURSOR
			, @cursor_created CURSOR
			, @running_count INT
			, @runnable_count INT
			, @failed_count INT
			, @unreachable_count INT
-- ============

-- ============
/*<code_comment><![CDATA[initialize output parameters]]></code_comment>*/
SET @task_group_run_id = NULL
-- ============

-- ============
/*<code_comment><![CDATA[check parameters]]></code_comment>*/
IF NULLIF(@group_type, N'') IS NULL
OR NULLIF(@group_name, N'') IS NULL
BEGIN
	RAISERROR('A required parameter is blank or null.', 16, 1)
	RETURN 1
END
-- ============

-- ============
/*<code_comment><![CDATA[set variables]]></code_comment>*/
SET @gbl_name_max_threads = N'PTX MAX THREADS (RUNTIME)'
SET @gbl_value_max_threads = ssf.gbl_get_variable_value(@gbl_name_max_threads)

SET @running_max_threads = ISNULL(@max_threads, ISNULL(@gbl_value_max_threads, 1))

IF @running_max_threads < 1
BEGIN
	RAISERROR('@max_threads must be 1 or greater.', 16, 1)
	RETURN 1
END

IF @loop_delay IS NULL
OR @loop_delay = '00:00:00'
	SET @loop_delay = '00:00:10.0'
SET @loop_delay_string = CAST(@loop_delay AS VARCHAR(12))

SET @log_directory = NULLIF(@log_directory, N'')
-- ============

-- ============
/*<code_comment><![CDATA[get task_group_id]]></code_comment>*/
SET @task_group_id = NULL
SET @task_group_id = ssf.ptx_get_task_group_id(@group_type, @group_name)

IF @task_group_id IS NULL
BEGIN
	RAISERROR('The task group (%s,%s) is not defined.', 16, 1, @group_type, @group_name)
	RETURN 1
END
-- ============

-- ============
/*<code_comment><![CDATA[validate task group dependencies]]></code_comment>*/
-- ============

-- ============
/*<code_comment><![CDATA[declare task group table variables]]></code_comment>*/
DECLARE @ptx_task TABLE (
	task_id INT NOT NULL PRIMARY KEY CLUSTERED
	, task_database_name NVARCHAR(128) NULL
	, task_procedure_name NVARCHAR(256) NOT NULL
	, task_parameter_list NVARCHAR(MAX) NULL
	, priority INT NOT NULL
	, semaphore_name NVARCHAR(128) NULL
);

DECLARE @ptx_task_dependency TABLE (
	task_dependency_id INT NOT NULL PRIMARY KEY CLUSTERED
	, task_id INT NOT NULL
	, depends_on_task_id INT NOT NULL
);

DECLARE @ptx_task_status TABLE (
	task_id INT NOT NULL PRIMARY KEY CLUSTERED
	, log_id INT NULL
	, job_id UNIQUEIDENTIFIER NULL
	, job_name NVARCHAR(128) NULL
	, eligible_fl BIT DEFAULT 0
	, unreachable_fl BIT DEFAULT 0
	, created_fl BIT DEFAULT 0
	, started_fl BIT DEFAULT 0
	, finished_fl BIT DEFAULT 0
	, failed_fl BIT DEFAULT NULL
);
-- ============

-- ============
/*<code_comment><![CDATA[declare cursor table variables]]></code_comment>*/
DECLARE @table_completed TABLE (
	task_id INT NOT NULL
	, log_id INT NULL
	, job_id UNIQUEIDENTIFIER NULL
	, job_name NVARCHAR(128) NULL
	, job_run_datetime DATETIME NULL
);

DECLARE @table_eligible TABLE (
	task_id INT NOT NULL
	, job_name NVARCHAR(128) NULL
);

DECLARE @table_created TABLE (
	task_id INT NOT NULL
	, job_id UNIQUEIDENTIFIER NULL
	, job_name NVARCHAR(128) NULL
);
-- ============

-- ============
/*<code_comment><![CDATA[get task list]]></code_comment>*/
SET @task = 'INSERT'
SET @task_description = 'Get list of active tasks for the task group'
	+ ' (' + @group_type + ',' + @group_name + ').'
SET @target_object = '@ptx_task'

EXECUTE ssf.log_begin
	@source_object = @source_object
	, @task = @task
	, @task_description = @task_description
	, @target_object = @target_object
	, @log_id = @log_id OUTPUT

INSERT @ptx_task
SELECT task_id
	, NULLIF(task_database_name, N'')
	, task_procedure_name
	, NULLIF(task_parameter_list, N'')
	, priority
	, NULLIF(semaphore_name, N'')
FROM ssf.ptx_get_active_task(@task_group_id)

SELECT @sql_error = @@ERROR
	, @row_count = @@ROWCOUNT

EXECUTE ssf.log_end
	@log_id = @log_id
	, @row_count = @row_count
	, @log_error_number = @sql_error

IF @sql_error <> 0
BEGIN
	RAISERROR('The INSERT into the table <@ptx_task> failed.', 16, 1)
	RETURN @sql_error
END

SET @task_count = @row_count
-- ============

-- ============
/*<code_comment><![CDATA[exit if no active tasks]]></code_comment>*/
IF @task_count = 0
	RETURN @@ERROR
-- ============

-- ============
/*<code_comment><![CDATA[get dependency list]]></code_comment>*/
SET @task = 'INSERT'
SET @task_description = 'Get list of active task dependencies for the task group'
	+ ' (' + @group_type + ',' + @group_name + ').'
SET @target_object = '@ptx_task_dependency'

EXECUTE ssf.log_begin
	@source_object = @source_object
	, @task = @task
	, @task_description = @task_description
	, @target_object = @target_object
	, @log_id = @log_id OUTPUT

INSERT @ptx_task_dependency
SELECT task_dependency_id
	, task_id
	, depends_on_task_id
FROM ssf.ptx_get_active_task_dependency(@task_group_id)

SELECT @sql_error = @@ERROR
	, @row_count = @@ROWCOUNT

EXECUTE ssf.log_end
	@log_id = @log_id
	, @row_count = @row_count
	, @log_error_number = @sql_error

IF @sql_error <> 0
BEGIN
	RAISERROR('The INSERT into the table <@ptx_task_dependency> failed.', 16, 1)
	RETURN @sql_error
END
-- ============

-- ============
/*<code_comment><![CDATA[make sure ptx job category exists]]></code_comment>*/
SET @job_category = N'[dynamic ptx job (Local)]'

EXECUTE @proc_error = ssf.job_add_job_category
	@category = @job_category

IF @proc_error <> 0
BEGIN
	RAISERROR('The call to the procedure <ssf.job_add_job_category> failed.', 16, 1)
	RETURN @proc_error
END
-- ============

-- ============
/*<code_comment><![CDATA[generate task group run id]]></code_comment>*/
SET @task_group_run_date = ssf.job_datetime_floor(GETDATE())

SET @task_group_run_id = ssf.ptx_generate_task_group_run_id(@group_type, @group_name, @task_group_run_date)
-- ============

-- ============
/*<code_comment><![CDATA[setup task status table]]></code_comment>*/
INSERT @ptx_task_status (
	task_id
	, job_name
)
SELECT task_id
	, ssf.ptx_generate_task_job_name(@task_group_id, task_id, @task_group_run_date)
FROM @ptx_task

SELECT @sql_error = @@ERROR
	, @row_count = @@ROWCOUNT

IF @sql_error <> 0
BEGIN
	RAISERROR('The INSERT into the table <@ptx_task_status> failed.', 16, 1)
	RETURN @sql_error
END
-- ============

-- ============
/*<code_comment><![CDATA[log beginning of task group execution]]></code_comment>*/
SET @task = 'RUN PTX TASK GROUP'
SET @task_description = @task_group_run_id + ': Run task group with <' + CAST(@running_max_threads AS VARCHAR(10)) + '> thread(s).'
SET @target_object = NULL

EXECUTE ssf.log_begin
	@source_object = @source_object
	, @task = @task
	, @task_description = @task_description
	, @target_object = @target_object
	, @log_id = @log_id OUTPUT

SET @task_group_log_id = @log_id
-- ============

-- ============
/*<code_comment><![CDATA[begin loop: manage task execution]]></code_comment>*/
SET @running_count = 0

WHILE 1 = 1
BEGIN

-- ============
/*<code_comment><![CDATA[if max threads parameter was not set, get current value of global variable]]></code_comment>*/
IF @max_threads IS NULL
BEGIN
	SET @gbl_value_max_threads = ISNULL(ssf.gbl_get_variable_value(@gbl_name_max_threads), 1)
	
	IF @gbl_value_max_threads < 1
	BEGIN
		RAISERROR('@max_threads must be 1 or greater.', 16, 1)
		RETURN 1
	END
	
	IF @gbl_value_max_threads <> @running_max_threads
	BEGIN
		SET @running_max_threads = @gbl_value_max_threads
		
		-- log change to max threads
		SET @task = 'RUN PTX TASK GROUP'
		SET @task_description = @task_group_run_id + ': Max threads has been updated to <' + CAST(@running_max_threads AS VARCHAR(10)) + '>.'
		SET @target_object = NULL

		EXECUTE ssf.log_begin
			@source_object = @source_object
			, @task = @task
			, @task_description = @task_description
			, @target_object = @target_object
			, @log_id = @log_id OUTPUT
		
		SET @sql_error = @@ERROR
		
		EXECUTE ssf.log_end
			@log_id = @log_id
			, @log_error_number = @sql_error
	END
END
-- ============

-- ============
/*<code_comment><![CDATA[begin cursor: for each newly completed job]]></code_comment>*/
DELETE @table_completed

INSERT @table_completed
SELECT task_id
	, log_id
	, job_id
	, job_name
	, ssf.job_get_first_run_datetime(job_id, @task_group_run_date)
FROM @ptx_task_status
WHERE started_fl = 1
AND finished_fl = 0
AND ssf.job_get_first_run_datetime(job_id, @task_group_run_date) IS NOT NULL

SET @cursor_completed = CURSOR FAST_FORWARD FOR
	SELECT c.task_id
		, c.log_id
		, c.job_id
		, c.job_name
		, c.job_run_datetime
		, t.semaphore_name
	FROM @table_completed c
	JOIN @ptx_task t
	ON c.task_id = t.task_id

OPEN @cursor_completed

FETCH NEXT FROM @cursor_completed INTO @task_id
	, @log_id
	, @job_id
	, @job_name
	, @job_run_datetime
	, @semaphore_name

WHILE @@FETCH_STATUS <> -1
BEGIN
	-- ============
	/*<code_comment><![CDATA[if holding semaphore connection, release it]]></code_comment>*/
	IF @semaphore_name IS NOT NULL
	BEGIN
		BEGIN TRY
			EXECUTE ssf.gbl_release_semaphore_connection
				@semaphore_name = @semaphore_name
				, @connections_to_release = 1
				, @locked_by_spid = @@SPID
				, @connections_released = @connections_released OUTPUT
		END TRY
		BEGIN CATCH
			-- how should error be handled here?
		END CATCH
	END
	-- ============

	-- ============
	/*<code_comment><![CDATA[decrement running task count]]></code_comment>*/
	SET @running_count -= 1

	IF @running_count < 0
		RAISERROR('Running task count is invalid, should never be less than zero.', 10, 1)
	-- ============

	-- ============
	/*<code_comment><![CDATA[get task job status]]></code_comment>*/
	SET @job_status = ssf.job_get_run_status(@job_id, @job_run_datetime)
	
	SET @failed_fl = CASE WHEN @job_status = 1 THEN 0 ELSE 1 END
	-- ============

	-- ============
	/*<code_comment><![CDATA[log end of task]]></code_comment>*/
	EXECUTE ssf.log_end
		@log_id = @log_id
		, @log_error_number = @failed_fl
	-- ============
	
	-- ============
	/*<code_comment><![CDATA[update task status - finished, failed]]></code_comment>*/
	UPDATE @ptx_task_status
	SET finished_fl = 1
		, failed_fl = @failed_fl
	WHERE task_id = @task_id
	-- ============

	-- ============
	/*<code_comment><![CDATA[delete job if it did not fail]]></code_comment>*/
	IF @failed_fl = 0
		EXECUTE ssf.job_delete
			@job_name = @job_name
	-- ============
	
	FETCH NEXT FROM @cursor_completed INTO @task_id
		, @log_id
		, @job_id
		, @job_name
		, @job_run_datetime
		, @semaphore_name
END
/*<code_comment><![CDATA[end cursor: for each newly completed job]]></code_comment>*/
-- ============

-- ============
/*<code_comment><![CDATA[update task status - unreachable]]></code_comment>*/
SET @row_count = 1

WHILE @row_count > 0
BEGIN
	UPDATE t
	SET t.unreachable_fl = 1
	FROM @ptx_task_status t
	WHERE t.unreachable_fl = 0
	AND t.started_fl = 0
	AND t.failed_fl IS NULL
	AND EXISTS (
		SELECT TOP 1 1
		FROM @ptx_task_dependency d
		JOIN @ptx_task_status dt
		ON d.depends_on_task_id = dt.task_id
		WHERE d.task_id = t.task_id
		AND (dt.failed_fl = 1 OR dt.unreachable_fl = 1)
	)

	SET @row_count = @@ROWCOUNT
END
-- ============

-- ============
/*<code_comment><![CDATA[update task status - eligible]]></code_comment>*/
UPDATE t
SET t.eligible_fl = 1
FROM @ptx_task_status t
WHERE eligible_fl = 0
AND unreachable_fl = 0
AND failed_fl IS NULL
AND NOT EXISTS (
	SELECT TOP 1 1
	FROM @ptx_task_dependency d
	LEFT JOIN @ptx_task_status dt
	ON d.depends_on_task_id = dt.task_id
	AND dt.finished_fl = 1
	AND dt.failed_fl = 0
	WHERE d.task_id = t.task_id
	AND dt.task_id IS NULL
)
-- ============

-- ============
/*<code_comment><![CDATA[begin cursor: for each eligible task not yet created]]></code_comment>*/
DELETE @table_eligible

INSERT @table_eligible
SELECT task_id
	, job_name
FROM @ptx_task_status
WHERE eligible_fl = 1
AND created_fl = 0
AND failed_fl IS NULL

SET @cursor_eligible = CURSOR FAST_FORWARD FOR
	SELECT s.task_id
		, s.job_name
		, t.task_database_name
		, t.task_procedure_name
		, t.task_parameter_list
	FROM @table_eligible s
	JOIN @ptx_task t
	ON s.task_id = t.task_id

OPEN @cursor_eligible

FETCH NEXT FROM @cursor_eligible INTO @task_id
	, @job_name
	, @task_database_name
	, @task_procedure_name
	, @task_parameter_list

WHILE @@FETCH_STATUS <> -1
BEGIN
	-- ============
	/*<code_comment><![CDATA[create task job]]></code_comment>*/
	BEGIN TRY
		IF @log_directory IS NOT NULL
			SET @log_file_path = @log_directory + '\' + @job_name + '.log'
		
		EXECUTE @proc_error = ssf.ptx_create_task_job
			@job_name = @job_name
			, @task_database_name = @task_database_name
			, @task_procedure_name = @task_procedure_name
			, @task_parameter_list = @task_parameter_list
			, @log_file_path = @log_file_path
			, @job_id = @job_id OUTPUT
	END TRY
	BEGIN CATCH
		SET @sql_error = ERROR_NUMBER()
	END CATCH
	
	SET @sql_error = ISNULL(NULLIF(@sql_error, 0), @proc_error)
	-- ============

	-- ============
	/*<code_comment><![CDATA[update task status - job_id, created, failed]]></code_comment>*/
	UPDATE @ptx_task_status
	SET job_id = CASE WHEN @sql_error = 0 THEN @job_id ELSE job_id END
		, created_fl = CASE WHEN @sql_error = 0 THEN 1 ELSE created_fl END
		, failed_fl = CASE WHEN @sql_error = 0 THEN failed_fl ELSE 1 END
	WHERE task_id = @task_id
	-- ============

	FETCH NEXT FROM @cursor_eligible INTO @task_id
		, @job_name
		, @task_database_name
		, @task_procedure_name
		, @task_parameter_list
END
/*<code_comment><![CDATA[end cursor: for each eligible task not yet created]]></code_comment>*/
-- ============

-- ============
/*<code_comment><![CDATA[get runnable task count]]></code_comment>*/
SET @runnable_count = (
	SELECT COUNT(*)
	FROM @ptx_task_status
	WHERE started_fl = 0
	AND unreachable_fl = 0
	AND failed_fl IS NULL
)
-- ============

-- ============
/*<code_comment><![CDATA[exit loop if no runnable or running tasks]]></code_comment>*/
IF @runnable_count = 0
AND @running_count = 0
	BREAK
-- ============

-- ============
/*<code_comment><![CDATA[begin cursor: for each created task not yet running, limited to available threads]]></code_comment>*/
DELETE @table_created

INSERT @table_created
SELECT task_id
	, job_id
	, job_name
FROM @ptx_task_status
WHERE created_fl = 1
AND started_fl = 0
AND failed_fl IS NULL

SET @cursor_created = CURSOR FAST_FORWARD FOR
	SELECT c.task_id
		, c.job_id
		, c.job_name
		, t.semaphore_name
	FROM @table_created c
	JOIN @ptx_task t
	ON c.task_id = t.task_id
	ORDER BY t.priority ASC
		, t.task_id ASC

OPEN @cursor_created

FETCH NEXT FROM @cursor_created INTO @task_id
	, @job_id
	, @job_name
	, @semaphore_name

WHILE @@FETCH_STATUS <> -1
AND @running_count < @running_max_threads
BEGIN

	-- ============
	/*<code_comment><![CDATA[begin: if semaphore connection required, acquire it]]></code_comment>*/
	IF @semaphore_name IS NOT NULL
	BEGIN
		BEGIN TRY
			EXECUTE ssf.gbl_acquire_semaphore_connection
				@semaphore_name = @semaphore_name
				, @connections_to_acquire = 1
				, @block_fl = 0
				, @connections_acquired = @connections_acquired OUTPUT
		END TRY
		BEGIN CATCH
			-- how should error be handled here?
		END CATCH
		
		-- ============
		/*<code_comment><![CDATA[if connection not acquired, skip to next task]]></code_comment>*/
		IF @connections_acquired < 1
		BEGIN
			FETCH NEXT FROM @cursor_created INTO @task_id
				, @job_id
				, @job_name
				, @semaphore_name
				
			CONTINUE
		END
		-- ============
	END
	/*<code_comment><![CDATA[end: if semaphore connection required, acquire it]]></code_comment>*/
	-- ============
	
	-- ============
	/*<code_comment><![CDATA[log start of task]]></code_comment>*/
	SET @task = 'RUN PTX TASK JOB'
	SET @task_description = @task_group_run_id + ': Run job for task_id <' + CAST(@task_id AS VARCHAR(10)) + '>.'
		+ '  Start job with <msdb.dbo.sp_start_job>; monitor history for job completion status.'
	SET @target_object = @job_name

	EXECUTE ssf.log_begin
		@source_object = @source_object
		, @task = @task
		, @task_description = @task_description
		, @target_object = @target_object
		, @log_id = @log_id OUTPUT
	-- ============

	-- ============
	/*<code_comment><![CDATA[start task job]]></code_comment>*/
	BEGIN TRY
		EXECUTE @proc_error = msdb.dbo.sp_start_job
			@job_id = @job_id
	END TRY
	BEGIN CATCH
		SET @sql_error = ERROR_NUMBER()
	END CATCH
	
	SET @sql_error = ISNULL(NULLIF(@sql_error, 0), @proc_error)
	-- ============

	IF @sql_error = 0
	BEGIN
		-- ============
		/*<code_comment><![CDATA[if successfully started, increment running task count]]></code_comment>*/
		SET @running_count += 1
		-- ============
	END
	ELSE
	BEGIN
		-- ============
		/*<code_comment><![CDATA[if failed to start, log end of task]]></code_comment>*/
		EXECUTE ssf.log_end
			@log_id = @log_id
			, @log_error_number = @sql_error
		-- ============
	END

	-- ============
	/*<code_comment><![CDATA[update task status - log_id, started, failed]]></code_comment>*/
	UPDATE @ptx_task_status
	SET log_id = CASE WHEN @sql_error = 0 THEN @log_id ELSE log_id END
		, started_fl = CASE WHEN @sql_error = 0 THEN 1 ELSE started_fl END
		, failed_fl= CASE WHEN @sql_error = 0 THEN failed_fl ELSE 1 END
	WHERE task_id = @task_id
	-- ============

	FETCH NEXT FROM @cursor_created INTO @task_id
		, @job_id
		, @job_name
		, @semaphore_name
END
/*<code_comment><![CDATA[end cursor: for each created task not yet running, limited to available threads]]></code_comment>*/
-- ============

-- ============
/*<code_comment><![CDATA[wait for loop delay - do not use processor constantly]]></code_comment>*/
IF @running_count > 0
	WAITFOR DELAY @loop_delay_string
-- ============

END
/*<code_comment><![CDATA[end loop: manage task execution]]></code_comment>*/
-- ============

-- ============
/*<code_comment><![CDATA[log end of task group execution]]></code_comment>*/
SET @failed_count = (
	SELECT COUNT(*)
	FROM @ptx_task_status
	WHERE failed_fl = 1
)

SET @unreachable_count = (
	SELECT COUNT(*)
	FROM @ptx_task_status
	WHERE unreachable_fl = 1
)

IF @failed_count > 0
OR @unreachable_count > 0
	SET @sql_error = 1

EXECUTE ssf.log_end
	@log_id = @task_group_log_id
	, @log_error_number = @sql_error

IF @sql_error <> 0
BEGIN
	SET @error_message = @task_group_run_id + ': Execution of the task group had'
		+ ' <' + CAST(@failed_count AS VARCHAR(10)) + '> failed tasks and'
		+ ' <' + CAST(@unreachable_count AS VARCHAR(10)) + '> unreachable tasks.'

	RAISERROR(@error_message, 16, 1)
	RETURN @sql_error
END
-- ============

RETURN @@ERROR
GO
